package com.chengjue.rocketmq.listener;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author: Kipeng Huang
 * @Date: 2018-4-12 15:50
 */
@Component
public class MessageListener implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            // TODO: 2018-4-12  消息处理
            for (MessageExt messageExt : msgs) {
                System.out.println("messageExt: " + messageExt);
                String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);
            }
        } catch (Exception e) {
            e.printStackTrace();
            //消费失败 稍后再试
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        //消费成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
